package org.apache.flink.kubernetes.kubeclient.factory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.flink.kubernetes.kubeclient.ComponentPod;
import org.apache.flink.kubernetes.kubeclient.decorators.CmdProviderManagerDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.ERConfMountDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.EnvSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.InitProviderManagerDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.KubernetesStepDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.MountSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesProviderManagerParameters;
import org.apache.flink.kubernetes.kubeclient.resources.KubernetesPod;
import org.apache.flink.util.Preconditions;

import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodBuilder;

/** Utility class for constructing the TaskManager Pod on the JobManager. */
public class KubernetesProviderManagerFactory {

    public static KubernetesPod buildTaskManagerKubernetesPod(
            ComponentPod podTemplate, KubernetesProviderManagerParameters kubernetesProviderManagerParameters) {
        ComponentPod componentPod = Preconditions.checkNotNull(podTemplate).copy();

        final List<KubernetesStepDecorator> stepDecorators =
                new ArrayList<>(
                        Arrays.asList(
                                new InitProviderManagerDecorator(kubernetesProviderManagerParameters),
                                new EnvSecretsDecorator(kubernetesProviderManagerParameters),
                                new MountSecretsDecorator(kubernetesProviderManagerParameters),
                                new CmdProviderManagerDecorator(kubernetesProviderManagerParameters)));

        stepDecorators.add(new ERConfMountDecorator(kubernetesProviderManagerParameters));

        for (KubernetesStepDecorator stepDecorator : stepDecorators) {
            componentPod = stepDecorator.decoratePod(componentPod);
        }

        final Pod resolvedPod =
                new PodBuilder(componentPod.getPodWithoutMainContainer())
                        .editOrNewSpec()
                        .addToContainers(componentPod.getMainContainer())
                        .endSpec()
                        .build();

        return new KubernetesPod(resolvedPod);
    }
}
